% Created 2020-03-15 Sun 18:19
% Intended LaTeX compiler: pdflatex
\documentclass[presentation]{beamer}
                                  \usepackage{xeCJK}
\setCJKmainfont{Source Han Serif CN}
\setCJKmonofont{JetBrains Mono}
\setmonofont{JetBrains Mono}
\usepackage[cache=false]{minted}
\usemintedstyle{rainbow_dash}
\usebackgroundtemplate{\includegraphics[width=\paperwidth,height=\paperheight]{../bg.jpg}}
\definecolor{BG_COLOR}{RGB}{1,137,255}
\setbeamercolor{frametitle}{bg=BG_COLOR}
\setminted{fontsize=\footnotesize}
\setbeamerfont{page number in head/foot}{size=\tiny}

                 \usepackage{default}
\usetheme{metropolis}
\date{2020-03-09}
\title{水滴异步架构实践}
\author{Jilen}
\subtitle{ 介绍非异步架构常见的问题及解决思路 }
\institute[Dripower]{水滴技术团队\\\url{https://scala.cool}}
\hypersetup{
 pdfauthor={Jilen \href{mailto:jilen.zhang@gmail.com}{Jilen} \texorpdfstring\{Jilen\newline\url{jilen.zhang@gmail.com}\}\{Jilen\} \%},
 pdftitle={水滴异步架构实践},
 pdfkeywords={},
 pdfsubject={},
 pdfcreator={Emacs 27.0.60 (Org mode 9.1.14)}, 
 pdflang={English}}
\begin{document}

\maketitle

\begin{frame}[label={sec:org8708840}]{本次主题}
\begin{itemize}
\item 全异步架构
\item 函数试编程
\item 并发问题
\item 流/批处理
\end{itemize}
\end{frame}
\begin{frame}[label={sec:org75777be}]{水滴技术栈}
\begin{itemize}
\item PlayFramework/cats-io/fs2
\item MySQL/PostgresSQL
\item Kafka/ES
\end{itemize}
\end{frame}

\begin{frame}[label={sec:orge7f30c3}]{常见的性能瓶颈}
\begin{itemize}
\item IO
\item Lock
\end{itemize}
\end{frame}

\begin{frame}[label={sec:org896e5bc}]{为什么选择 Scala}
\begin{itemize}
\item 基于 JVM 有良好生态
\item 用 twitter 这样的(成功?)案例
\item 性能不错、尤其是全异步之后
\end{itemize}
\end{frame}

\begin{frame}[label={sec:org3afaa22}]{IO 阻塞带来的问题}
\begin{itemize}
\item 提高吞吐量 -> 增加线程数
\item 增加线程数 -> 增加线程切换开销 -> 降低吞吐量
\end{itemize}
\end{frame}

\begin{frame}[label={sec:org525baf1}]{早期的容器阻塞模型}
\begin{center}
\includegraphics[width=.9\linewidth]{images/container-pool.pdf}
\end{center}
\end{frame}

\begin{frame}[label={sec:orgbe37bbe}]{PlayFramework 非阻塞模型}
\begin{center}
\includegraphics[width=.9\linewidth]{images/play-pool.pdf}
\end{center}
\end{frame}

\begin{frame}[label={sec:orgcc2c83b}]{PlayFramework 优势}
\begin{itemize}
\item 简单：核心（Router）就是 `Request => Future[Result]` 的（偏）函数
\item 核心全异步的实现
\end{itemize}
\end{frame}

\begin{frame}[label={sec:org561dca2}]{异步数据库访问}
PostgresAsync
\begin{itemize}
\item 基于 Netty
\item 有较多的应用案例
\item 没人维护了（我司fork了一个，自己维护）
\end{itemize}

Finagle
\begin{itemize}
\item 基于 Netty
\item Postgres 尚不成熟
\end{itemize}

Jasyncsql
\begin{itemize}
\item 从 PostgresAsync 派生，使用 kotlin 重写
\item 目前有商业公司在维护
\end{itemize}
\end{frame}

\begin{frame}[fragile,label={sec:org67cf0b7}]{异步操作组合（回调形式）}
 \begin{minted}[]{java}
public void <A> findExists(String name, callback: (User) -> A)
public void <A> createUser(User user, callback: (long) -> A)

public void handle(Request request, Response response) {
  String username = ...;
  findExists(username, (User u) -> {
    if(u != null) {
      response.set("NameExists");
    } else {
      createUser(u, (long id) -> {
        response.set(id.toString);
      });
    }
  });
}
\end{minted}
\end{frame}

\begin{frame}[fragile,label={sec:orga014241}]{异步操作组合（Future/Promise）}
 \begin{minted}[]{java}
public CompletableFuture<User> findExists(String name) {}
public CompletableFuture<Long> createUser(User user) {}
public CompletableFuture<Result>  handle(Request r) {
  String name = ...;
  findExists(name).thenCompose((User u) -> {
    if(u != null) {
      CompletableFuture
        .completedFuture(Result.of("NameExists"));
    } else {
      createUser(u).thenApply((Long id) -> {
        Result.of(id.toString);
      });
    }
  });
}
\end{minted}
\end{frame}

\begin{frame}[fragile,label={sec:orga13a591}]{异步组合 Monad Composition}
 \begin{minted}[]{scala}
def findExists(name: String): Future[Option[User]]
def createUser(user: user): Future[Long]
def createOrReportError(ou: Some[User]) = ou match {
  case Some(u) =>
    Future.successful(Result.of("NameExists"))
  case None =>
    createUser(user)
}
def handle(r: Request): Future[Result] = {
  val name = ...
  for {
   ou <- findExists(name)
   r <- createOrReportError(ou)
  } yield r
}
\end{minted}
\end{frame}

\begin{frame}[fragile,label={sec:org965cad5}]{异步组合 协程（Async/Await)}
 \begin{minted}[]{kotlin}
suspend fun findExists(name: String): User? {}
suspend fun createUser(user: User): Long {}

suspend fun handle(r: Request): Future[Result] {
  val name = ...
  u = await(findExists(name))
  if(u != null)
    return Result.of("NameExists")
  else {
   val id = await(createUser(user))
   return Results.of(id.toString())
  }
}

\end{minted}
\end{frame}

\begin{frame}[fragile,label={sec:org722959f}]{Kotlin 协程背后}
 CPS 变换
\begin{minted}[]{java}
fun findExists(name: String, callback: Contiuation<User?>) {}
interface Contiuation<A> {
  fun resumeWith(result: Result<T>)
}
\end{minted}
\end{frame}

\begin{frame}[label={sec:orgb123de4}]{函数试编程}
\begin{itemize}
\item 声明式
\item 引用透明
\end{itemize}
\end{frame}
\begin{frame}[fragile,label={sec:orgab3ada9}]{为什么引用透明如此重要}
 \begin{minted}[]{kotlin}
val yesterdayDTL =
  LocalDate.now().minusDays(1).astStartOfDay();

fun fun0(): List<Data> {
  LocalDateTime dt = LocalDate.now()
      .minusDays(1)
      .astStartOfDay();
  val dtl = readFromDB(dt);
  ...
}

fun fun1() {
  LocalDateTime dt = LocalDate.now()
      .minusDays(1)
      .astStartOfDay();
  val dtl = readFromDB(dt);
  ...
}

\end{minted}
\end{frame}

\begin{frame}[fragile,label={sec:org4ebf646}]{IO Monad 隔离副作用}
 \begin{minted}[]{kotlin}
sealed class IO<A> {
  internal data class Succeed<V>(val v: V) : IO<V>()
  internal data class FlatMap<V1, V2>(val f: (A1) -> IO<V2>): IO<V2>()
  internal data class Effect<V>(val f: () -> IO<V>): IO<V>()
  fun <B> flatmap(f: (A) -> B): IO<B> = FlatMap(this, f)
}
fun runIO(io: IO<A>): A
object Console {
  fun getStr(): IO<String> = IO.effect(() => readLine())
  fun putStrLn(str: String): IO<Unit> = IO.effect(() => println(str))
}
fun plus(i1: IO<Int>, i2: IO<Int>): IO<Int> {
  return i1.flatMap { ii ->
   i2.map { jj -> ii + jj }
  }
}
\end{minted}
\end{frame}


\begin{frame}[label={sec:org1dcb374}]{高并发的挑战}
\begin{itemize}
\item 数据正确
\item 高性能
\end{itemize}
\end{frame}

\begin{frame}[label={sec:org53ad4c9}]{锁的问题}
\begin{itemize}
\item 锁不可怕，竞争激烈的锁才可怕
\item 激烈的竞争会导致锁升级，进一步加剧竞争
\item 标准库yy （OpenJDK/Hotspot） 有很多常见的隐藏锁，如加载类
\end{itemize}
\end{frame}

\begin{frame}[fragile,label={sec:org0bc91e1}]{一个例子}
 \begin{minted}[]{java}
public void service(Request req, Response resp) {
  return  Class
    .forName("xHandlerClass")
    .getConstructor()
    .newInstance().handle(req, resp);
}
\end{minted}
\end{frame}


\begin{frame}[label={sec:orgb31941a}]{一个实际案例}
减库存问题
\end{frame}

\begin{frame}[fragile,label={sec:org637d61e}]{数据库悲观锁}
 \begin{minted}[]{sql}
UPDATE item
  SET stock = stock - 1, modified_at = NOW()
  WHERE id = $1 AND stock > 0;
\end{minted}
竞争激烈时性能奇差
\end{frame}


\begin{frame}[fragile,label={sec:org7177825}]{数据库乐观锁}
 \begin{minted}[]{sql}
UPDATE item
  SET stock = stock - 1, modified_at = NOW()
  WHERE id = $1 AND stock = $2;
\end{minted}
竞争激烈导致操作失败率非常高
\end{frame}

\begin{frame}[label={sec:orgdbb78c2}]{一个方案}
\begin{itemize}
\item 生产者消费者模式
\item 批量处理
\end{itemize}
\end{frame}

\begin{frame}[label={sec:orgd7ddb49}]{请求入队}
\begin{center}
\includegraphics[width=.9\linewidth]{images/queue.pdf}
\end{center}
\end{frame}

\begin{frame}[fragile,label={sec:org17d5769}]{一个细节}
 结果返回给用户问题

\begin{minted}[]{java}
private final Queue<Pair<Request,
  CompletableFuture<Result>>> queue = ...;
public CompletableFuture<Result> accept(Request r) {
  CompletableFuture res =  new CompletableFuture<Result>();
  if(!queue.tryEnqueue(new Pair(r, res))) {
    res.completeExceptionally(new Exception("Busy!"));
  }
  return res;
}
public void process() {
  while(true) {
    Pair<Request, CompletableFuture<Result> r = queue.dequeue();
    long id = save(r._1);
    r._2.complete(Result.of(id));
  }
}

\end{minted}
\end{frame}

\begin{frame}[fragile,label={sec:orgcb91423}]{Cats-IO / fs2 的方案}
 \begin{minted}[]{scala}
val queue: Queue[F[_], (Request, Deferred[F, Result])]
def accept(r: Request) =
  Deferred[F, Result].flatMap(d => queue.enqueue(r -> d))
def process() = queue.dequeue.chunks.evalMap { chunks =>
  save(chunks.map(_._1)).flatMap { ids =>
    (chunks.map(_._2) zip ids).traverse {
      case (d, id) => d.complete(id)
    }
  }
}
\end{minted}
\end{frame}

\begin{frame}[label={sec:org06a0013}]{CQRS (Akka Persistence) 方案}
\end{frame}
\end{document}